package org.dh2580.delay.queue.bean.bean;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.sql.DataSource;

import org.dh2580.delay.queue.bean.AbstractMsgListener;
import org.dh2580.delay.queue.bean.db.DBUtil;
import org.dh2580.delay.queue.bean.executor.SimpleExecutor;
import org.dh2580.delay.queue.bean.executor.SimpleScheduleExecutor;
import org.dh2580.delay.queue.bean.task.ClearMsgTask;
import org.dh2580.delay.queue.bean.task.SelectExpiredMsgTask;
import org.dh2580.delay.queue.bean.task.SelectNewMsgTask;

/**
 * @Author: hao.deng
 * @Date: 2019/5/14-下午7:17
 */
public class DelayQueueBean {

    public static final int ABANDON_THRESHOLD_DAY = 1;

    private DataSource dataSource;

    private SimpleScheduleExecutor selectExecutor;

    private SimpleExecutor listenerExecutor;

    private SimpleExecutor callbackExecutor;

    private Integer corePoolSize;

    private Integer maxPoolSize;

    private Integer selectIntervalSec;

    private Integer maxExpiredMinute;

    private List<AbstractMsgListener> listeners;

    public DelayQueueBean(DataSource dataSource, Integer corePoolSize, Integer maxPoolSize, Integer selectIntervalSec, Integer maxExpiredMinute) {
        this.dataSource = dataSource;

        this.corePoolSize = corePoolSize;
        this.maxPoolSize = maxPoolSize;

        this.selectIntervalSec = selectIntervalSec;
        this.maxExpiredMinute = maxExpiredMinute;
    }

    public void setListeners(List<AbstractMsgListener> listeners) {
        this.listeners = listeners;
    }

    public void registeListner(AbstractMsgListener listener) {
        if (this.listeners == null) {
            synchronized (this) {
                if (this.listeners == null) {
                    this.listeners = new ArrayList<>(32);
                }
            }
        }

        this.listeners.add(listener);
    }

    public void init() {
        DBUtil.setDataSource(this.dataSource);

        this.selectExecutor = new SimpleScheduleExecutor(3, "delay-queue-select");
        this.listenerExecutor = new SimpleExecutor(this.corePoolSize, this.maxPoolSize, "delay-queue-listener");
        this.callbackExecutor = new SimpleExecutor(5, 10, "delay-queue-callback");

        Map<String, List<AbstractMsgListener>> routeKeyListeners = new HashMap<>();
        if (listenerExecutor != null) {
            for (AbstractMsgListener listener : listeners) {
                if (!routeKeyListeners.containsKey(listener.getRouteKey())) {
                    routeKeyListeners.put(listener.getRouteKey(), new ArrayList<>(32));
                }

                routeKeyListeners.get(listener.getRouteKey()).add(listener);
            }
        }


        SelectNewMsgTask newMsgTask = new SelectNewMsgTask(this.listenerExecutor, this.callbackExecutor, routeKeyListeners, this.maxExpiredMinute);
        SelectExpiredMsgTask expiredMsgTask = new SelectExpiredMsgTask(this.listenerExecutor, this.callbackExecutor, routeKeyListeners, this.maxExpiredMinute);

        selectExecutor.scheduleAtFixedRate(newMsgTask, 0, this.selectIntervalSec, TimeUnit.SECONDS);
        selectExecutor.scheduleAtFixedRate(expiredMsgTask, 0, this.maxExpiredMinute, TimeUnit.MINUTES);

        selectExecutor.scheduleAtFixedRate(new ClearMsgTask(), 0, 1, TimeUnit.HOURS);


    }
}
